package mr

import (
	"fmt"
	"hash/fnv"
	"io"
	"log"
	"net/rpc"
	"os"
	"sort"
	"strings"
)

// Map functions return a slice of KeyValue.
type KeyValue struct {
	Key   string
	Value string
}

type ByKey []KeyValue

// for sorting by key.
func (a ByKey) Len() int           { return len(a) }
func (a ByKey) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a ByKey) Less(i, j int) bool { return a[i].Key < a[j].Key }

// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}

func DoReduceTask(reducef func(string, []string) string, task *RpcTask) {
	sourceFile, err := os.Open(task.SourceFile)
	if err != nil {
		fmt.Printf("reduce err:%v\n", err)
	}
	defer sourceFile.Close()
	content, err := io.ReadAll(sourceFile)
	if err != nil {
		fmt.Printf("reduce err:%v\n", err)
	}
	kva := []KeyValue{}
	for _, v := range strings.Split(string(content), "\n") {
		oneLine := strings.Split(v, " ")
		if len(oneLine) != 2 {
			fmt.Printf("err line: %v\n", v)
			continue
		}
		kva = append(kva, KeyValue{Key: oneLine[0], Value: oneLine[1]})
	}

	if len(kva) == 0 {
		fmt.Printf("%v nothing\n", task.SourceFile)
		return
	}
	sort.Sort(ByKey(kva))
	oname := fmt.Sprintf("mr-out-%v", task.SourceFile)
	ofile, _ := os.Create(oname)
	i := 0
	for i < len(kva) {
		j := i + 1
		for j < len(kva) && kva[j].Key == kva[i].Key {
			j++
		}
		values := []string{}
		for k := i; k < j; k++ {
			values = append(values, kva[k].Value)
		}
		output := reducef(kva[i].Key, values)

		// this is the correct format for each line of Reduce output.
		fmt.Fprintf(ofile, "%v %v\n", kva[i].Key, output)

		i = j
	}
	ofile.Close()
}

func DoMapTask(mapf func(string, string) []KeyValue, task *RpcTask) {
	//fmt.Println("start map work")
	file, err := os.Open(task.SourceFile)
	if err != nil {
		fmt.Printf("cannot open %v,err: %v\n", task.SourceFile, err)
		return
	}
	content, err := io.ReadAll(file)
	if err != nil {
		fmt.Printf("cannot read %v", task.SourceFile)
		return
	}
	file.Close()
	kva := mapf(task.SourceFile, string(content))
	for _, v := range kva {
		targetIndex := fmt.Sprintf("%v", ihash(v.Key)%task.NReduce)
		outFile, err := os.OpenFile(targetIndex, os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0644)
		if err != nil {
			fmt.Printf("map err: %v\n", err)
		}
		fmt.Fprintf(outFile, "%v %v\n", v.Key, v.Value)
	}
}

// main/mrworker.go calls this function.
func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {
	lastTask := RpcTask{
		TaskType: INIT_TASK,
	}
	nowTask := RpcTask{}

	// Your worker implementation here.
	CallGetNewTask(&lastTask, &nowTask)
	lastTask = nowTask
	for nowTask.TaskType != STOP_WORK_TASK {
		if nowTask.TaskType == MAP_TASK {
			//fmt.Printf("map task:%v start\n", nowTask.SourceFile)
			DoMapTask(mapf, &nowTask)
		} else if nowTask.TaskType == REDUCE_TASK {
			//fmt.Printf("reduce task:%v start\n", nowTask.SourceFile)
			DoReduceTask(reducef, &nowTask)
		}
		if !CallGetNewTask(&lastTask, &nowTask) {
			nowTask.TaskType = STOP_WORK_TASK
		}
		lastTask = nowTask
	}
	//fmt.Printf("worker end\n")
}

func CallGetNewTask(lastTask *RpcTask, nowTask *RpcTask) bool {
	fmt.Printf("send: lastTask:%v;nowTask:%v\n", lastTask, nowTask)
	ok := call("Coordinator.GetNewTask", lastTask, nowTask)
	if ok {
		fmt.Printf("receive: lastTask:%v;nowTask:%v\n", lastTask, nowTask)
		return true
	} else {
		fmt.Println("fail")
		return false
	}
}

// example function to show how to make an RPC call to the coordinator.
//
// the RPC argument and reply types are defined in rpc.go.
func CallExample() {

	// declare an argument structure.
	args := ExampleArgs{}

	// fill in the argument(s).
	args.X = 99

	// declare a reply structure.
	reply := ExampleReply{}

	// send the RPC request, wait for the reply.
	// the "Coordinator.Example" tells the
	// receiving server that we'd like to call
	// the Example() method of struct Coordinator.
	ok := call("Coordinator.Example", &args, &reply)
	if ok {
		// reply.Y should be 100.
		fmt.Printf("reply.Y %v\n", reply.Y)
	} else {
		fmt.Printf("call failed!\n")
	}
}

// send an RPC request to the coordinator, wait for the response.
// usually returns true.
// returns false if something goes wrong.
func call(rpcname string, args interface{}, reply interface{}) bool {
	// c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
	sockname := coordinatorSock()
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		log.Fatal("dialing:", err)
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}

	fmt.Println(err)
	return false
}
